/*
 * Copyright 2011-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0.
 * See `LICENSE` in the project root for license information.
 */

package me.ijleex.mgmt.framework.concurrent.distlock.redis;

import java.util.concurrent.ExecutionException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.lettuce.core.ScriptOutputType;
import io.lettuce.core.SetArgs;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;

/**
 * 基于 Redis 的分布式锁
 *
 * @author liym
 * @since 2019-04-11 16:49 新建
 */
public final class DistributedLock {

    /**
     * 分布式锁标识前缀
     *
     * @since 2019-04-11 18:39
     */
    private static final String LOCK_PREFIX = "distlock:";

    private static final String LOCK_OK = "OK";

    private static final Long UNLOCK_OK = 1L;

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * 异步 Redis-Cluster API
     *
     * @since 2019-04-11 16:59
     */
    private final RedisClusterAsyncCommands<String, String> redisCommands;

    private final String lockKey;

    /**
     * 构造分布式锁
     *
     * @param conn 有状态的 Redis-Cluster 连接
     * @param key 锁定变量的名称或类型等，如 product（产品）、user（用户）等
     */
    public DistributedLock(StatefulRedisClusterConnection<String, String> conn, String key) {
        /// Assert.notNull(conn, "conn must not be null");
        this.redisCommands = conn.async();
        this.lockKey = LOCK_PREFIX + key;
    }

    /**
     * 获取分布式锁
     *
     * @param x 锁定变量的值，如线程ID、商品ID等
     * @param timeout 锁的超期时间（秒）
     * @since 2020-01-13 17:21
     */
    public void lock(String x, long timeout) {
        while (true) {
            if (this.tryLock(x, timeout)) {
                return;
            }
        }
    }

    /**
     * 尝试获取分布式锁
     *
     * @param x 锁定变量的值，如线程ID、商品ID等
     * @param timeout 锁的超期时间（秒）
     * @return 加锁是否成功：true/false
     * @since 2019-04-11 17:17
     */
    public boolean tryLock(String x, long timeout) {
        SetArgs setArgs = SetArgs.Builder.ex(timeout).nx();

        String result;
        try {
            result = this.redisCommands.set(this.lockKey, x, setArgs).get();
        } catch (InterruptedException | ExecutionException e) {
            result = null;
            this.logger.error("tryLock error", e);
        }

        return LOCK_OK.equals(result);
    }

    /**
     * 释放分布式锁
     *
     * @param x 锁定变量的值，如线程ID、商品ID等
     * @return 解锁是否成功：true/false
     * @since 2019-04-11 17:33
     */
    public boolean unlock(String x) {
        // 使用 Lua 脚本，保证操作的 `原子性`
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

        String[] keys = new String[]{this.lockKey};

        Object result;
        try {
            result = this.redisCommands.eval(script, ScriptOutputType.INTEGER, keys, x).get();
        } catch (InterruptedException | ExecutionException e) {
            result = 0L;
            this.logger.error("unlock error", e);
        }

        return UNLOCK_OK.equals(result);
    }

    /**
     * 判断锁是否被持有
     *
     * @param x 锁定变量的值，如线程ID、商品ID等
     * @return 锁是否被持有：true/false
     * @since 2019-04-12 10:32
     */
    public boolean isLocked(String x) {
        // 使用 Lua 脚本，保证操作的 `原子性`
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return 1 else return 0 end";

        String[] keys = new String[]{this.lockKey};

        Object result;
        try {
            result = this.redisCommands.eval(script, ScriptOutputType.INTEGER, keys, x).get();
        } catch (InterruptedException | ExecutionException e) {
            result = 0L;
            this.logger.error("isLocked error", e);
        }

        return UNLOCK_OK.equals(result);
    }

}
